package com.cupdata.sms.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.protocol.types.Field;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @author: zda
 * @date: 2024/5/6 11:16
 */
@Component
@Slf4j
public class KafkaSendTemplate {
    @Autowired
    private KafkaProducer<String,byte[]> kafkaProducer;
    public void send(String topic,byte[] message){
        send(topic,null,message);
    }
    public void send(String topic,Integer partition,byte[] message){
        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
        send(topic,uuid,partition,message);
    }
    public void sendByKey(String topic,String key,byte[] message){
        send(topic,key,null,message);
    }
    public void send( String topic,String key,Integer partition,byte[] message){
        ProducerRecord<String, byte[]> stringProducerRecord = new ProducerRecord<>(topic,partition, key, message);
        kafkaProducer.send(stringProducerRecord,((metadata, exception) -> {
            if(exception!=null){
                log.error("发送消息失败：topic:{},key:{},异常：{}",topic,key,new String(message));
            }else {
                log.info("发送消息成功：topic:{},key:{},partition：{},offset：{}",topic,key,metadata.partition(),metadata.offset());
            }
        }));
    }



}
